package org.example.sql;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import static org.apache.flink.table.api.Expressions.$;

public class SqlDemo {
    public static void main(String[] args) throws Exception {
        //todo 1 创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //todo 2创建表
        //1.1 写法一创建配置对象
//        EnvironmentSettings settings = EnvironmentSettings
//                .newInstance()
//                .inStreamingMode()
//                .build();
//        TableEnvironment.create(settings);

        //1.2 写法二
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        tableEnv.executeSql("CREATE TABLE source(\n" +
                "\tid INT,\n" +
                "\tts BIGINT,\n" +
                "\tvc INT\n" +
                ") WITH (\n" +
                "  'connector'='datagen',\n" +
                "  'rows-per-second'='1',\n" +
                "  'fields.id.kind'='random',\n" +
                "  'fields.id.min'='1',\n" +
                "  'fields.id.max'='10',\n" +
                "  'fields.ts.kind'='sequence',\n" +
                "  'fields.ts.start'='1',\n" +
                "  'fields.ts.end'='1000000',\n" +
                "  'fields.vc.kind'='random',\n" +
                "  'fields.vc.min'='1',\n" +
                "  'fields.vc.max'='100'\n" +
                ");");

        tableEnv.executeSql("CREATE TABLE sink(\n" +
                "\tid INT,\n" +
                "\tsumVC INT\n" +
                ") WITH (\n" +
                "  'connector'='print'\n" +
                "  );");

        //todo 3 执行查询
        //table查询
        //Table table = tableEnv.sqlQuery("select id,sum(vc) as sumVC from source where id>5 group by id  ;");
        //把table对象注册为表名
        //tableEnv.createTemporaryView("tmp", table);

        // tableEnv.executeSql("select * from tmp;");

        //table api查询
//        Table source = tableEnv.from("source");
//        Table result = source
//                .where($("id").isGreater(5))
//                .groupBy($("id"))
//                .aggregate($("vc").sum().as("sumVC"))
//                .select($("id"), $("sumVC"));

        //todo 4 输出
        tableEnv.executeSql("INSERT INTO sink SELECT id, sumVC FROM (SELECT id, SUM(vc) as sumVC FROM source WHERE id > 5 GROUP BY id)");

        // 执行 Flink 作业
        //result.executeInsert("sink");
        //env.execute("SqlDemo");
    }

}
